package zhongbo.javaweb.instagram.util.thread;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import zhongbo.javaweb.instagram.util.ArrayUtil;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 说明：
 *
 * @author zhongboWork
 * @version 1.0
 * @since 2021-12-17 11:37
 */
@Slf4j
@Data
public abstract class ProgressTask<T> implements Observer {

    private String taskName;

    private List<ProgressRunner<T>> progressRunner;

    /**
     * 储存当前正在运行的任务 (id, taskRunner)
     */
    private Map<String, ProgressRunner<T>> runningTask;

    private boolean isRunning;

    private boolean isPaused;

    private int threadSize;

    private Queue<T> targetList;

    private List<T> finishedList;

    private Queue<Queue<T>> partitionList;

    private ProgressProcessor<T> processor;

    private AtomicInteger runningCount = new AtomicInteger();

    private ThreadPoolExecutor poolExecutor;

    public ProgressTask(String taskName, int threadSize, Queue<T> targetList, ProgressProcessor<T> processor) {
        this.taskName = taskName;
        this.threadSize = threadSize;
        this.targetList = targetList;
        this.processor = processor;
    }

    public void start() {
        synchronized (this) {
            if (!this.isRunning) {
                this.isRunning = true;
                try {
                    log.info("启动任务 {}", taskName);
                    initProgressTask();
                    for (ProgressRunner<T> tProgressRunner : progressRunner) {
                        poolExecutor.execute(tProgressRunner);
                        String progressName = tProgressRunner.getProgressName();
                        runningTask.put(progressName, tProgressRunner);
                    }
                    updateTask();
                } catch (Exception e) {
                    stop();
                    String msg = String.format("启动任务 %s 失败!", taskName);
                    throw new RuntimeException(msg, e);
                }
            }
        }
    }

    public void stop() {
        this.isRunning = false;
        for (ProgressRunner<T> tProgressRunner : progressRunner) {
            tProgressRunner.stop();
        }
        updateTask();
    }

    public void pause() {
        for (ProgressRunner<T> tProgressRunner : progressRunner) {
            tProgressRunner.pause();
        }
        this.isPaused = true;
        updateTask();
    }

    public void resume() {
        for (ProgressRunner<T> tProgressRunner : progressRunner) {
            tProgressRunner.resume();
        }
        this.isPaused = false;
    }

    @Override
    public void update(Observable o, Object arg) {
        if (o instanceof ProgressRunner) {
            ProgressMessage<T> message = (ProgressMessage<T>) arg;
            if (message == null) {
                String msg = "程序代码错误! ProgressRunner 未传递 ProgressMessage 参数信息";
                log.error(msg, new RuntimeException(msg));
                return;
            }
            if (!message.isFinished()) {
                T data = message.getData();
                if (data != null) {
                    finishedList.add(data);
                } else {
                    String msg = "程序代码错误! ProgressRunner 未传递 ProgressMessage data 参数信息";
                    log.error(msg, new RuntimeException(msg));
                }
            } else {
                ProgressRunner<?> progressRunner = (ProgressRunner<?>) o;
                String progressName = progressRunner.getProgressName();
                poolExecutor.remove(progressRunner);
                runningTask.remove(progressName);
                int runningCount = this.runningCount.decrementAndGet();
                log.info("子进程 {} 结束运行, 剩余运行中子进程 {} 个", progressName, runningCount);
                if (runningCount == 0) {
                    log.info("程序执行完毕! 剩余运行中子进程 {} 个", runningCount);
                }
                this.isPaused = false;
                this.isRunning = false;
            }
            updateTask();
        }
    }

    protected abstract void updateTask();

    protected void initProgressTask() {
        if (targetList == null) {
            throw new RuntimeException("ProgressTask 使用错误,未指定 targetList!");
        }
        if (processor == null) {
            throw new RuntimeException("ProgressTask 使用错误,未指定 processor!");
        }
        if (targetList.size() == 0) {
            throw new RuntimeException("ProgressTask 使用错误,数据为空 targetList!");
        }
        this.finishedList = new ArrayList<>();
        List<List<T>> lists = partitionTaskStrategy(new ArrayList<>(targetList), threadSize);
        Queue<Queue<T>> tQueue = new ConcurrentLinkedQueue<>();
        for (List<T> list : lists) {
            tQueue.add(new ConcurrentLinkedQueue<>(list));
        }
        this.partitionList = new ConcurrentLinkedQueue<>(tQueue);
        this.runningTask = new ConcurrentHashMap<>(tableSizeFor(targetList.size()));
        this.progressRunner = new ArrayList<>();
        Iterator<Queue<T>> iterator = partitionList.iterator();
        for (int i = 0; i < partitionList.size(); i++) {
            Queue<T> ts = iterator.next();
            ProgressRunner<T> progressRunner = new ProgressRunner<>(ts, buildProgressRunnerName(i), this.processor);
            this.progressRunner.add(progressRunner);
        }

        ThreadFactory threadFactory = new ThreadFactory() {
            private final ThreadFactory defaultFactory = Executors.defaultThreadFactory();
            private final AtomicInteger threadNumber = new AtomicInteger(1);

            @Override
            public Thread newThread(Runnable r) {
                Thread thread = this.defaultFactory.newThread(r);
                if (!thread.isDaemon()) {
                    // 守护线程;
                    // 当JVM中所有的线程都是守护线程的时候，JVM就可以退出了；如果还有一个或以上的非守护线程则JVM不会退出。
                    thread.setDaemon(true);
                }

                thread.setName("ProgressTask-" + taskName + "-Runner-" + this.threadNumber.getAndIncrement());
                return thread;
            }
        };
        int queueSize = Math.max(threadSize, partitionList.size());
        poolExecutor = new ThreadPoolExecutor(threadSize, threadSize, 10, TimeUnit.SECONDS, new ArrayBlockingQueue<>(queueSize), threadFactory);

    }

    protected String buildProgressRunnerName(int index) {
        return String.format("%s-sub-%d", this.taskName, index);
    }

    /**
     * 任务拆分策略
     * <p>
     * 可以@Override自定义
     *
     * @param targetList     任务目标列表
     * @param partitionSizes 分块数量
     * @return 拆分后数组
     */
    private List<List<T>> partitionTaskStrategy(List<T> targetList, int partitionSizes) {
        return ArrayUtil.divAverageList(targetList, partitionSizes);
    }

    /**
     * 返回2的次方倍数
     *
     * @param c 整数
     * @return 2的次方数
     */
    private int tableSizeFor(int c) {
        int n = c - 1;
        n |= n >>> 1;
        n |= n >>> 2;
        n |= n >>> 4;
        n |= n >>> 8;
        n |= n >>> 16;
        return (n < 0) ? 1 : n + 1;
    }


}
